package com.dtkavin.spark.demo

import java.io.{BufferedReader, File, FileReader}

import org.apache.spark.{SparkContext, SparkConf}

import scala.collection.mutable.ArrayBuffer

/**
  * Created by IntelliJ IDEA.
  * Programmer : John Zn
  * Date : 2016/4/13 0013
  * Time : 10:41
  * Discribtion : 
  */
class BinarySearch {
  //  val ipMap=new util.LinkedHashMap[String,Array[String]]()
  val arrBuf = new ArrayBuffer[String]()

  def ip2Long(ip: String): Long = {
    val fragments = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until fragments.length) {
      ipNum = fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }

  def ip2L(ip: String): Long = {
    val ipArr = ip.split("\\.")
    var ipNum = 0L
    for (i <- 0 until ipArr.length) {
      ipNum = ipArr(i).toLong | ipNum << 8L
    }
    ipNum
  }

  //单机版 导入ip文件
  def loadIPs(): Unit = {
    val file = new File("D:\\CZ\\FQDownload\\feiq\\Recv Files\\ip.txt")
    val reader = new BufferedReader(new FileReader(file))
    var str: String = ""
    while ( {
      str = reader.readLine();
      str
    } != null) {
      arrBuf += str
    }
    println("load done")
  }

  //单机版 二分法查找
  def stand_alone_BinSearch(ipNum: Long): String = {
    var low = 0
    var high = arrBuf.length - 1
    var res: String = ""
    while (low <= high) {
      println(s"$low : $high")
      if (ipNum >= arrBuf((low + high) / 2).split("\\|")(2).toLong && ipNum <= arrBuf((low + high) / 2).split("\\|")(3).toLong) {
        return arrBuf((low + high) / 2)
      }
      if (ipNum > arrBuf((low + high) / 2).split("\\|")(3).toLong) {
        low = (low + high) / 2 + 1
      } else if (ipNum < arrBuf((low + high) / 2).split("\\|")(2).toLong)
        high = (low + high) / 2 - 1
    }
    "-1"
  }

  def binSearch(ips: Array[(Long, Long, String)], ipNum: Long): Long = {
    var low = 0
    var high = ips.length - 1
    var middle = 0

    while (low <= high) {
//      println(s" $low : $high")
      middle = (low + high) / 2
      if (ipNum >= ips(middle)._1 && ipNum <= ips(middle)._2)
        return middle
      if (ipNum > ips(middle)._2) {
        low = middle + 1
      } else if (ipNum < ips(middle)._1) {
        high = middle - 1
      }
    }
    -1
  }
}

object stand_alone {
  def main(args: Array[String]) {
    val ip: String = "115.120.17.80"
    val bs = new BinarySearch
    val ipNum = bs.ip2Long(ip)
    println(s"ipNum: $ipNum")
    bs.loadIPs()
    println(bs.stand_alone_BinSearch(ipNum))
  }
}

object BinarySearch {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("UsrPassBy").setMaster("local[3]")
    val sc = new SparkContext(conf);

    val ips = sc.textFile("D:\\CZ\\FQDownload\\feiq\\Recv Files\\ip.txt").map {
      x => {
        val line = x.split("\\|")
        //            1.0.1.0|1.0.3.255|16777472|16778239|亚洲|中国|福建|福州||电信|350100|China|CN|119.306239|26.075302
        (line(2).toLong, line(3).toLong, line(6))
      }
    }.collect

    val bs = new BinarySearch

    val ip: String = "115.120.17.80"
    val ipNum = bs.ip2L(ip)
    val offset = bs.binSearch(ips, ipNum)
    println(s"ipNum : $ipNum")
    println(s"offset : $offset")
    println(ips(offset.toInt))


    sc.stop()
  }
}

object Test extends App{
  val ip: String = "115.120.17.80"
  ip.split("[.]").foreach(println(_))
}